package services

import (
	"encoding/json"
	"fmt"
	"github.com/zw2582/ginlib"
	"go.uber.org/zap"
	"reflect"
	"runtime/debug"
	"strings"
	"sync"
	"szxf/models"
	"time"
)

var (
	dbQueueServiceDefault *dbQueueService
	once sync.Once
)

//实例化一个DbQueueService
func NewDbQueueService() *dbQueueService {
	if dbQueueServiceDefault == nil {
		once.Do(func() {
			key := "queue:db:app_name:"+ginlib.APP_NAME
			dbQueueServiceDefault = &dbQueueService{
				queue:ginlib.NewRedisMq(key, ginlib.RedisCli),
			}
		})
	}
	return dbQueueServiceDefault
}

//dbQueueService 数据异步操作服务
type dbQueueService struct {
	queue ginlib.RedisMq
}

//插入需要异步消化的数据库对象
func (this *dbQueueService) Push(dbObj models.DBQueueInterface) {
	raw, err := json.Marshal(dbObj)
	if err != nil {
		ginlib.Logger.Error("dbQueueService push fail:"+err.Error(), zap.Any("dbobj", dbObj))
	}
	tableName := models.DB.TableName(dbObj)
	this.queue.PushMessage(tableName+"|"+string(raw))
}

//Consume 消费数据库对象
func (this *dbQueueService) Consume() {
	taskPool := ginlib.NewTaskPool(10, 0)
	for {
		msg := this.queue.GetMessage()
		if msg == "" {
			time.Sleep(time.Second)
			continue
		}
		//执行操作
		taskPool.Do(msg, func(i interface{}) {
			msg = i.(string)
			if err := this.consumeDo(msg); err != nil {
				ginlib.Logger.Error("dbQueueService Consume Fail:"+err.Error(), zap.String("msg", msg), zap.Error(err))
			} else {
				//应答消息
				this.queue.MessageAck(msg)
			}
		})
	}
}

//consumeDo 实际数据库消费，存储redis队列中的msg到数据库
func (this *dbQueueService) consumeDo(msg string) (err error) {
	defer func() {
		if e := recover(); e != nil {
			switch val := e.(type) {
			case error:
				err = val
			default:
				err = fmt.Errorf("%+v", err)
			}
			ginlib.Logger.Error(err.Error(), zap.String("Stack", string(debug.Stack())))
		}
	}()

	//解析json数据
	msgs := strings.Split(msg, "|")
	tableName := msgs[0]
	raw := msgs[1]

	//遍历所有数据库对象
	for _, val := range ginlib.XormGetSyncObjs() {
		if models.DB.TableName(val) == tableName {
			var valObj interface{}
			if reType := reflect.TypeOf(val); reType.Kind() == reflect.Ptr {
				//如果是个指针则先转化为结构,并以此新建结构指针
				valObj = reflect.New(reType.Elem()).Interface()
			}
			if err = json.Unmarshal([]byte(raw), valObj); err != nil {
				return
			}
			//保存到数据库
			ginlib.Logger.Debug("保存到数据库:"+tableName, zap.String("raw:", raw), zap.Any("valObj:", valObj))
			if tmp, ok := valObj.(models.DBQueueInterface); ok {
				tmp.Save()
			} else {
				ginlib.Logger.Error("is not DBQueueInterface")
			}
		}
	}


	return nil
}

